package com.rxjava.simpleflatmap;

/**
 * 被观察者，用来生产发送事件；
 */
public class Observable<T> {

    private OnSubscribe<T> onSubscribe;

    private Observable(OnSubscribe<T> onSubscribe) {
    	//onSubscribe -> subscriber.onNext(6);
    	//onSubscribe -> OnSubscribeLift对象(OnSubscribeLift对象)
    	//onSubscribe -> OnSubscribeFromArray对象
        this.onSubscribe = onSubscribe;
    }

    public final void subscribe(Subscriber<? super T> subscriber) {
    	System.out.println("step1");
    	//subscriber -> System.out.println(s);
    	onSubscribe.call(subscriber);
    }

    public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
    	//onSubscribe -> subscriber.onNext(6);
    	//onSubscribe -> OnSubscribeFromArray对象
        return new Observable<>(onSubscribe);
    }

    public static <T> Observable<T> just(final T t1, final T t2) {
        return from((T[]) new Object[]{t1, t2});
    }

    public static <T> Observable<T> from(T[] array) {
    	//array -> new String[]{"flatMap" + 6, "flatMap" + 7, "flatMap" + 8}
        return create(new OnSubscribeFromArray<T>(array));
    }

    public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
    	//func -> return Observable.from(new String[]{"flatMap" + integer, "flatMap" + 7, "flatMap" + 8});
    	//map(func) -> Observable对象(OnSubscribeLift对象)
        return merge(map(func));
    }

    public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
    	//func -> return Observable.from(new String[]{"flatMap" + integer, "flatMap" + 7, "flatMap" + 8});
        return lift(new OperatorMap<T, R>(func));
    }

    public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
    	//operator -> OperatorMap对象
    	//operator -> OperatorMerge对象
        return new Observable<>(new OnSubscribeLift<T, R>(onSubscribe, operator));
    }

    private static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
    	//source -> Observable对象(OnSubscribeLift对象)
        return source.lift(new OperatorMerge<T>());
    }

}